package org.kth.dks.test_bed;

/**
 * <p>Title: </p>
 *
 * <p>Description: DKS middleware</p>
 *
 * <p>Copyright: Copyright (c) 2005</p>
 *
 * <p>Company: </p>
 *
 * @author Ali Ghodsi (aligh@kth.se)
 * @version 1.0
 */

import org.kth.dks.DKSCallbackInterface;
import org.kth.dks.DKSImpl;
import org.kth.dks.DKSObject;
import org.kth.dks.dks_comm.ConnectionManager;
import org.kth.dks.dks_comm.DKSOverlayAddress;
import org.kth.dks.dks_node.DKSNode;
import org.kth.dks.dks_node.DefaultAppHandler;

public class TestMany {

    public static void main(String[] args) {
        try {
            int NODES = 20;
            ConnectionManager cm = ConnectionManager.getInstance(4440);
            DKSImpl[] nodes = new DKSImpl[NODES];
            for (int i=0; i<NODES; i++) {
                long x = i*10;
                DKSOverlayAddress oa = new DKSOverlayAddress("dksoverlay://0/"+x+"/0");
                nodes[i] = new DKSImpl(cm, oa);
                if (i==0) nodes[0].create();
                else nodes[i].join( nodes[0].getDKSRef() );
            }
            sleep(1000);

            for (int i=0; i<NODES; i++) {
              System.out.println(nodes[i].getNodeInfo().successor+ " " + nodes[i].getNodeInfo().status );
              DefaultAppHandler ah = new DefaultAppHandler() {
                public void broadcastCallback(DKSObject payload){
                  DKSNode n = (DKSNode) dks;
//                  if (payload.getData().length==3 && payload.getData()[0]==1 && payload.getData()[1]==2 && payload.getData()[2]==3)
//                    System.err.println("RIGHT DATA "+n.getMyDKSRef());
//                  else
//                    System.err.println("WRONG DATA "+n.getMyDKSRef());
                }
              };
              DKSCallbackInterface c = nodes[i].setCallbackHandler( ah );
              ah.setDKSCallbackInterface(c);

              nodes[i].logLevel(2);
            }

            sleep(1000);


            nodes[0].broadcastRestricted(new DKSObject(new byte[]{1,2,3}), 100, 161);

            sleep(2000);

            for (int i=0; i<NODES; i++) {
                nodes[i].leave();
                sleep(100);
            }
            cm.end();

        } catch( Exception ex ) {
            ex.printStackTrace();
        }
    }

    public static void sleep(long msec) {
        try {
            Thread.sleep(msec);
        } catch (Exception ex) { ex.printStackTrace(); }
    }

}
